package mr.dividends;

import java.io.IOException;

import mr.dividends.model.DividendsInputWritable;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
import au.com.bytecode.opencsv.CSVParser;

public class DividendsWritablesRecordReader extends RecordReader<LongWritable, DividendsInputWritable>{
    private final LineRecordReader lineReader = new LineRecordReader();
    private final DividendsInputWritable dividends = new DividendsInputWritable();
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        lineReader.initialize(split, context);
        lineReader.nextKeyValue();
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        boolean hasNextKey = lineReader.nextKeyValue();
        if (hasNextKey){
            parse(lineReader.getCurrentValue());
        }
        return hasNextKey;
    }
    
    private void parse(Text value) throws IOException {
    	CSVParser csvParser = new CSVParser();
    	String[] split = csvParser.parseLine(value.toString());
        dividends.setExchange(split[0]);
        dividends.setStockSymbol(split[1]);
        dividends.setDateStr(split[2]);
        dividends.setDividends(Float.parseFloat(split[3]));
    }

    @Override
    public LongWritable getCurrentKey() throws IOException, InterruptedException {
        return lineReader.getCurrentKey();
    }

    @Override
    public DividendsInputWritable getCurrentValue() throws IOException, InterruptedException {
        return dividends;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return lineReader.getProgress();
    }

    @Override
    public void close() throws IOException {
        lineReader.close();
    }
}